
package rendezvousqueue;

import harness.Queue;
import harness.Random;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

import sun.misc.Unsafe;
import java.lang.reflect.Field;
import java.util.Collection;
import java.util.Iterator;

public class RendezvousSyncQueue implements Queue {
    static final private Object FREE_NODE = new Object();

    static final private int MAX_THREADS = Runtime.getRuntime().availableProcessors();
    static final private int MAX_RING_SIZE = MAX_THREADS;
    static final private int MAX_RING_SIZE_1 = (MAX_RING_SIZE - 1);

    boolean stop_ds_flag = false;

    private volatile int _ring_size = 0;

    // for AtomicInteger
    private static final Unsafe unsafe = getUnsafe();

    private static Unsafe getUnsafe() {
        try {
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            return (Unsafe) field.get(null);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public Object poll(int threadId, Random rng) {
        return get(threadId, rng.nextInt(MAX_THREADS));
    }

    public boolean offer(Object e, int threadId, Random rng) {
        return put(e, threadId, rng.nextInt(MAX_THREADS));
    }

    // Used for the publication list
    static class RingNode {
        final int index;

        public static final long prodOffset;
        public static final long consOffset;

        static {
            try {
                prodOffset = unsafe.objectFieldOffset(RingNode.class.getDeclaredField("_prod_item"));
                consOffset = unsafe.objectFieldOffset(RingNode.class.getDeclaredField("_cons_item"));
            } catch (Exception ex) {
                throw new Error(ex);
            }
        }

        RingNode () {
            this.index = -1;
        }

        RingNode (int index) {
            this.index = index;
        }

        volatile RingNode		next = null;
        // Padding to separate index and next from the item references.
        Object pad1, pad2, pad3, pad4, pad5, pad6, pad7, pad8, pad9, pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17, pad18, pad19, pad20, pad21, pad22, pad23, pad24, pad25, pad26, pad27;

        volatile Object _cons_item = FREE_NODE;
        volatile Object _prod_item = FREE_NODE;

        // Paddings after all fields to prevent false sharing.
        Object opad1, opad2, opad3, opad4, opad5, opad6, opad7, opad8, opad9, opad10, opad11, opad12, opad13, opad14, opad15, opad16, opad17, opad18, opad19, opad20, opad21, opad22, opad23, opad24, opad25, opad26, opad27, opad28, opad29, opad30;
    }

    final protected static AtomicReferenceFieldUpdater<RingNode, RingNode> nextUpdater =
        AtomicReferenceFieldUpdater.newUpdater(RingNode.class, RingNode.class, "next");

    private final RingNode[] _node_ptrs;
    private final RingNode _head;

    public Object get(int thread, int rand_thread_id) {
        int ring_size = ringSize();
        int ctr = 0;

        mainLoop:
        while (true) {
            int hashId = hash (rand_thread_id, ring_size);
            RingNode l_node = _node_ptrs[hashId];
            Object cur_obj;
            int busy_ctr = 0;

            do {
                cur_obj = l_node._cons_item;
                if (cur_obj == FREE_NODE) {
                    if (unsafe.compareAndSwapObject (l_node, RingNode.consOffset, FREE_NODE, null))
                        break;
                }

                l_node = l_node.next;

                if (++busy_ctr >= ring_size) {
                    increaseRing(_node_ptrs[ringSize()]);
                    ring_size = ringSize();
                    busy_ctr = 0;
                }
            } while (!stop_ds_flag);

                // spin until the node points to a producer
            while ((cur_obj = l_node._cons_item) == null) {

                if ((++ctr & 63) == 0) {
                    if ((l_node.index > (ring_size = ringSize())) && unsafe.compareAndSwapObject (l_node, RingNode.consOffset, null, FREE_NODE)) {
                        // find myself a new node
                        continue mainLoop;
                    }

                }

                if (stop_ds_flag)
                    return null;
            }

            l_node._cons_item = FREE_NODE;

            if (busy_ctr < 2 && ctr > 64)
                decreaseRing(_node_ptrs[ring_size]);

            return cur_obj;
        }
    }


    public boolean put(Object value, int thread, final int rand_thread_id) {

        int ring_size = ringSize();
        int hashId = hash (rand_thread_id, ring_size);

        RingNode local_iter = _node_ptrs[hashId];
        RingNode iter = local_iter.next;
        RingNode casing_iter;
        Object curr_obj = null;
        int ctr = 0;

        do {
            if (local_iter._cons_item == null)
                casing_iter = local_iter;
            else {
                casing_iter = iter;
                iter = iter.next;
            }

            curr_obj = casing_iter._cons_item;
            if (curr_obj == null) {
                if (unsafe.compareAndSwapObject (casing_iter, RingNode.consOffset, null, value))
                    return true;
                ctr = 0;
            }

            ctr++;
         } while (!stop_ds_flag);

        return false;
    }

    // constructor
   public RendezvousSyncQueue() {
        stop_ds_flag = false;

        RingNode p_node = null, p_prev = null;

        // create the ring
        _node_ptrs = new RingNode[MAX_THREADS];
        for (int i = 0; i < _node_ptrs.length; i++) {
            p_node = new RingNode (i);
            if (p_prev != null) {
                p_node.next = p_prev;
            }
            p_prev = p_node;

            _node_ptrs[i] = p_node;
        }

        _node_ptrs[0].next = _node_ptrs[1]; // the list is cyclic
        _head = _node_ptrs[0];
    }

    protected void increaseRing (RingNode tail) {
        int ring_size = tail.index;

        if (ring_size < _node_ptrs.length - 1) {
            nextUpdater.compareAndSet(_head, tail, _node_ptrs[ring_size + 1]);
        }

    }

    protected void decreaseRing (RingNode tail) {
        int ring_size = tail.index;

        if (ring_size == 0) {
            // can't decrease less than 0
            return;
        }

        nextUpdater.compareAndSet(_head, tail, _node_ptrs[ring_size - 1]);
    }

    protected final int ringSize() {
        return _head.next.index;
    }

    public String name() {
        return this.getClass().toString();
    }

    public void stop_ds() {
        stop_ds_flag = true;
    }

    static protected final int hash (long id, final int ring_size) {
        return (int)id % (ring_size + 1);
    }

    public int size() {
        //stop_ds_flag = false;
        return ringSize ();
    }

}
